package broker

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/golang/protobuf/proto"
	"go-micro.dev/v4/broker"
	"manager/bll"
	"manager/client/grpc"
	log "manager/collector/logger"
	"manager/event"
	"manager/lib/trace_log"
	"manager/model"
	SignalPo "manager/model/signal"
	"manager/store/redis"
	"manager/utils"
	"manager/utils/goroutine_pool"
	"proto/dto"
)

var topics = []string{
	"signal.event.status",
	"signal.event.location",
	"signal.event.alarm",
	"signal.event.fault",
	"signal.event.control",
	"signal.event.battery",
	"signal.event.signal",
	"event.device.fault.check",
	"signal.event.ota",
}

type consumer struct {
	pubSub broker.Broker
}

func NewConsumer(pubSub broker.Broker) *consumer {
	return &consumer{
		pubSub: pubSub,
	}
}

func init() {
	event.Register(func(data *event.Data) {
		DeviceCyclingReport.onEvent(data)
		CommandHandler.onEvent(data)
		Consumer.onEvent(data)
	})
}

// Init 初始化
func (c *consumer) Init() error {
	for i := range topics {
		c.pubSub.Subscribe(topics[i], c.Handle)
	}
	return nil
}

func (c *consumer) init() func() {
	return func() {}
}

func (c *consumer) onEvent(data *event.Data) {
	if data == nil || (data.Type != event.Event_Mq_SyncDevice) {
		return
	}

	obj := data.Data.(*event.MqSyncDevice)
	if obj == nil {
		log.Error("consumer onEvent err")
		return
	}

	c.SyncDeviceInfoToShadow(obj.Sn, obj.UpdateType, obj.MapValue)
	return
}

// Handle 处理器
func (c *consumer) Handle(eventData broker.Event) error {
	defer utils.ExceptionCatch()
	body := eventData.Message().Body
	switch eventData.Topic() {
	// 状态上报
	case "signal.event.status":
		// 状态上报
		goroutine_pool.GetPoolV2("status").Push(context.Background(), func(d interface{}) error {
			defer utils.ExceptionCatch()
			data := &dto.DeviceStatus{}
			err := proto.Unmarshal(body, data)
			if err != nil {
				log.Errorf("consumer handle Unmarshal err:%v", err)
				return err
			}

			// 获取bikeId
			bikeId, err := bll.CacheBike.GetBikeIdBySn(data.Sn)
			if err != nil {
				log.Errorf("consumer handle GetBikeIdBySnCode err:%v", err)
				return err
			}

			// 打印日志
			trace_log.DebugLogger(data.Sn, "----------- ", data.Event.String(), " ----------------")

			// 落库状态数据
			c.deviceStatusHandler(data.Sn, bikeId, data.Event)

			// 上下电
			if data.Event == dto.EventType_PowerOn || data.Event == dto.EventType_PowerOff {
				// 加分布式锁
				conn := redis.KV.GetRedisPool().Get()
				lock := &redis.TRedLock{}
				bSuccess, _ := lock.Lock(conn, fmt.Sprintf("onoff_%v", data.Sn), 10)
				if !bSuccess {
					log.Errorf("consumer off on lock fail")
					return nil
				}
				defer lock.UnLock(conn)

				// 上电
				if data.Event == dto.EventType_PowerOn {
					trace_log.DebugLogger(data.Sn, "车辆上电")
					event.Emit(&event.Data{
						Type: event.Event_Power_On,
						Data: &event.BikePowerOn{EbikeId: bikeId},
					})

					/*toggle := int32(1)
					// 车辆上电更新车辆配置
					event.Emit(&event.Data{
						Type: event.Event_Bike_Cmd,
						Data: &event.BikeCmd{
							EbikeId: bikeId,
							CmdType: "toggle",
							Body: &dto.SwitchRequest{
								Sn:                  data.Sn,
								PreventOutOfControl: &toggle,
							},
						},
					})*/
				}

				// 下电
				if data.Event == dto.EventType_PowerOff {
					trace_log.DebugLogger(data.Sn, "车辆下电")

					event.Emit(&event.Data{
						Type: event.Event_Power_Off,
						Data: &event.BikePowerOff{EbikeId: bikeId},
					})

					// 车辆下电
					isOn, _ := bll.DeviceStatus.GetPowerOnUser(bikeId)
					if !isOn {
						trace_log.DebugLogger(data.Sn, "consumer EventType_PowerOff 车辆已经下电")
						return nil
					}

					// 查询位置
					goroutine_pool.GetPool().Push(nil, func(params interface{}) error {
						lo, _ := bll.SignalV2.GetMsgFromDevice("", &SignalPo.GetMessageReq{
							EbikeId: bikeId,
							Cmd:     "Location",
						})

						if lo != nil {
							point, _ := json.Marshal(lo)
							Consumer.SyncDeviceInfoToShadow(data.Sn, dto.EnUpdateType_ATTRIBUTE, map[string]interface{}{
								"stop_location": string(point),
							})
						}
						return nil
					})
				}
			}

			// 上线
			if data.Event == dto.EventType_Online {
				trace_log.DebugLogger(data.Sn, "车辆上线")
				defer func() {
					trace_log.DebugLogger(data.Sn, "车辆上线完成")
				}()
				redis.KV.AddHash(context.Background(), "online_device", data.Sn, "1")
				CornTask.AddOnlineDevice(data.Sn)
				event.Emit(&event.Data{
					Type: event.Event_Bike_Online,
					Data: &event.BikeOnline{BikeId: bikeId},
				})
			}

			// 下线
			if data.Event == dto.EventType_Offline {
				trace_log.DebugLogger(data.Sn, "车辆下线")
				defer func() {
					trace_log.DebugLogger(data.Sn, "车辆下线完成")
				}()
				redis.KV.DelHash(context.Background(), "online_device", data.Sn)
				trace_log.RemoveLogger(data.Sn)
				CornTask.RemoveOnlineDevice(data.Sn)
				event.Emit(&event.Data{
					Type: event.Event_Bike_Offline,
					Data: &event.BikeOffline{BikeId: bikeId},
				})
			}

			return nil
		})
	// 位置上报
	case "signal.event.location":
		// 落库位置数据
		goroutine_pool.GetPool().Push(nil, func(d interface{}) error {
			defer utils.ExceptionCatch()
			data := &dto.Location{}
			err := proto.Unmarshal(body, data)
			if err != nil {
				log.Errorf("consumer handle Unmarshal err:%v", err)
				return err
			}

			// 是否存在
			if !CornTask.IsExist(data.Sn) {
				CornTask.AddOnlineDevice(data.Sn)
			}

			// 位置上报
			if err = DeviceCyclingReport.LocationReport(data); err != nil {
				log.Errorf("consumer handle LocationReport err:%v", err)
				return err
			}
			return nil
		})
	// 故障上报
	case "signal.event.fault":
		// 故障上报
		goroutine_pool.GetPoolV2("fault").Push(nil, func(d interface{}) error {
			defer utils.ExceptionCatch()
			data := &dto.Fault{}
			err := proto.Unmarshal(body, data)
			if err != nil {
				log.Errorf("consumer handle Unmarshal err:%v", err)
				return err
			}

			if err = FaultReport.HandlerFaultReport(data); err != nil {
				log.Errorf("consumer handle HandlerFaultReport err:%v", err)
			}
			return nil
		})
	case "event.device.fault.check":
		// 故障上报
		goroutine_pool.GetPoolV2("fault").Push(nil, func(d interface{}) error {
			defer utils.ExceptionCatch()
			data := &dto.FaultCheck{}
			err := proto.Unmarshal(body, data)
			if err != nil {
				log.Errorf("consumer handle Unmarshal err:%v", err)
				return err
			}

			if err = FaultReport.HandlerFaultCheckReport(data); err != nil {
				log.Errorf("consumer handle HandlerFaultCheckReport err:%v", err)
			}
			return nil
		})
	// 告警
	case "signal.event.alarm":
		// 异步处理警告
		goroutine_pool.GetPoolV2("alarm").Push(nil, func(d interface{}) error {
			defer utils.ExceptionCatch()
			data := &dto.Alarm{}
			err := proto.Unmarshal(body, data)
			if err != nil {
				log.Errorf("consumer handle Unmarshal err:%v", err)
				return err
			}

			// 告警上报
			if err = AlarmReport.HandlerAlarm(data); err != nil {
				log.Errorf("consumer handle HandlerAlarm err:%v", err)
				return err
			}
			return nil
		})
	case "signal.event.control":
		goroutine_pool.GetPoolV2("control").Push(nil, func(d interface{}) error {
			defer utils.ExceptionCatch()
			data := &dto.Controller{}
			err := proto.Unmarshal(body, data)
			if err != nil {
				log.Errorf("consumer handle control err:%v", err)
				return err
			}

			if data.FirmwareVersion == "" {
				trace_log.ErrorLogger.Info("中空版本上报 consumer handle control FirmwareVersion is empty sn:", data.Sn)
				log.Errorf("consumer handle control FirmwareVersion is empty sn:%s", data.Sn)
				return nil
			}

			// 更新设备中控版本
			trace_log.DebugLogger(data.Sn, "中控版本上报 consumer handle control FirmwareVersion:", data.FirmwareVersion)
			bikeInfo, _ := bll.CacheBike.GetBikeInfoBySn(data.Sn)
			if bikeInfo != nil {
				// 更新设备中控版本
				bll.Control.Update(context.Background(), &model.ControlUpdateRequest{
					Id:              bikeInfo.ControlId,
					FirmwareVersion: &data.FirmwareVersion,
				})
			}

			// 同步设备信息到设备影子
			c.SyncDeviceInfoToShadow(data.Sn, dto.EnUpdateType_ATTRIBUTE, map[string]interface{}{
				"firmware_version": data.FirmwareVersion,
			})

			// 中控上报
			if err = ControlReport.HandlerControl(data); err != nil {
				log.Errorf("consumer handle HandlerControl err:%v", err)
				return err
			}

			return nil
		})
	case "signal.event.battery":
		goroutine_pool.GetPoolV2("battery").Push(nil, func(d interface{}) error {
			defer utils.ExceptionCatch()
			data := &dto.BatteryInfo{}
			err := proto.Unmarshal(body, data)
			if err != nil {
				log.Errorf("consumer handle battery err:%v", err)
				return err
			}

			// 同步设备信息到设备影子
			c.SyncDeviceInfoToShadow(data.Sn, dto.EnUpdateType_ATTRIBUTE, map[string]interface{}{
				"status":  int32(data.Status),
				"mileage": data.Mileage,
			})
			return nil
		})
	case "signal.event.signal":
		goroutine_pool.GetPoolV2("signal").Push(nil, func(d interface{}) error {
			defer utils.ExceptionCatch()
			data := &dto.Signal{}
			err := proto.Unmarshal(body, data)
			if err != nil {
				log.Errorf("consumer handle Signal err:%v", err)
				return err
			}

			// 同步设备信息到设备影子
			c.SyncDeviceInfoToShadow(data.Sn, dto.EnUpdateType_ATTRIBUTE, map[string]interface{}{
				"signal_4g":  fmt.Sprintf("%v", data.Signal_4G),
				"signal_gps": fmt.Sprintf("%v", data.SignalGps),
			})
			return nil
		})
	case "signal.event.ota":
		goroutine_pool.GetPoolV2("ota").Push(nil, func(d interface{}) error {
			defer utils.ExceptionCatch()
			data := &dto.OTAUpgradeNtf{}
			err := proto.Unmarshal(body, data)
			if err != nil {
				log.Errorf("consumer OTAUpgradeNtf handle Signal err:%v", err)
				return err
			}

			trace_log.DebugLogger(data.Sn, "OTAUpgradeNtf received:", utils.ToJsonString(data))
			// 成功
			if data.Result == 0 {
				//查询设备影子中的版本信息
				vk := &dto.GetShadowAttributeKeysResponse{}
				vk, err = grpc.SCmd.GetShadowAttributeKeys(context.Background(), &dto.GetShadowAttributeKeysRequest{
					DeviceName: data.Sn,
					Key:        "firmware_version",
				})
				if err != nil {
					trace_log.DebugLogger(data.Sn, "OTAUpgradeNtf GetShadowAttributeKeys firmware_version err:", err)
					return err
				}

				if vk == nil || len(vk.Value) == 0 {
					trace_log.DebugLogger(data.Sn, fmt.Sprintf("No version information found,sn:%v", data.Sn))
					return fmt.Errorf(fmt.Sprintf("No version information found,sn:%v", data.Sn))
				}
				err = bll.Ota.OTAUpgradeSuccess(data.Sn, vk.Value)
				if err != nil {
					trace_log.DebugLogger(data.Sn, "OTAUpgradeNtf OTAUpgradeSuccess err:", err)
				}
				return nil
			}

			err = bll.Ota.OTAUpgradeFailed(data.Sn, bll.Ota.GetInfo(int(data.Result)))
			if err != nil {
				trace_log.DebugLogger(data.Sn, "OTAUpgradeNtf OTAUpgradeSuccess err:", err)
			}
			return nil
		})
	}
	return fmt.Errorf("no handler")
}

// SyncDeviceInfoToShadow 同步设备状态
func (c *consumer) SyncDeviceInfoToShadow(sn string, updateType dto.EnUpdateType, mapValue map[string]interface{}) error {
	updateRequest := &dto.DoUpdateRequest{
		DeviceName: sn,
		Type:       updateType,
	}

	for k, v := range mapValue {
		updateRequest.Items = append(updateRequest.Items, &dto.Item{
			Key:   k,
			Value: fmt.Sprintf("%v", v),
		})
	}

	if len(mapValue) == 0 {
		return fmt.Errorf("no update value")
	}

	// 序列化
	sendMsg, err := proto.Marshal(updateRequest)
	if err != nil {
		return err
	}

	// 发布消息
	err = PubSub.Publish("shadow.event.report", &broker.Message{
		Header: map[string]string{
			"method": "update",
		},
		Body: sendMsg,
	})

	return err
}

// deviceStatusHandler 设备状态
func (c *consumer) deviceStatusHandler(sn string, bikeId int64, eventType dto.EventType) error {
	status := 0
	statusKey := ""
	switch eventType {
	case dto.EventType_PowerOn:
		statusKey = "power"
		status = 1
	case dto.EventType_PowerOff:
		statusKey = "power"
		status = 0
	case dto.EventType_Online:
		statusKey = "online"
		status = 1
	case dto.EventType_Offline:
		statusKey = "online"
		status = 0
	case dto.EventType_Lock:
		statusKey = "lock"
		status = 1
	case dto.EventType_Unlock:
		statusKey = "lock"
		status = 0
	}
	if statusKey == "" {
		return nil
	}

	// 设备状态落库
	trace_log.DebugLogger(sn, "设备状态落库 ", statusKey, ":", status)
	redis.KV.AddHash(context.Background(), fmt.Sprintf("device_status:%v", sn), statusKey, status)

	// 发射命令
	goroutine_pool.GetPool().Push(nil, func(d interface{}) error {
		// 同步设备状态
		if err := c.SyncDeviceInfoToShadow(sn, dto.EnUpdateType_STATUS, map[string]interface{}{
			statusKey: status,
		}); err != nil {
			log.Errorf("syncDeviceStatus error:%v", err)
		}

		// 命令上报
		event.Emit(&event.Data{
			Type: event.Event_Command_Report,
			Data: &event.CommandReport{
				EbikeId: bikeId,
				Cmd:     eventType,
			},
		})
		return nil
	})

	// 解除锁定，解除失窃上报
	if statusKey == "lock" && status == 0 {
		err := bll.LoseLock.RemoveLoseLock(sn)
		if err != nil {
			trace_log.DebugLogger(sn, "consumer 解除失窃失败")
		}
	}
	return nil
}
